package com.quanyan.ic.middleware.mq.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by zhuo.shi on 2017/8/21.
 * 修改为抽象类
 */
public abstract class AbstractProducer {

    private static final Logger logger = LoggerFactory.getLogger(AbstractProducer.class);

    private DefaultMQProducer producer;
    private String producerGroup;
    private String nameServerAddress;
    private String instanceName;

    public void start() {
        this.nameServerAddress = getNameServerAddress();
        this.instanceName = getInstanceName();
        this.producerGroup = getProducerGroup();
        producer = this.getProducer();//保证应用内唯一
        producer.setNamesrvAddr(nameServerAddress);
            /*默认情况下不需要设置instanceName，rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示
             如果同一个jvm中，不同的producer需要往不同的rocketmq集群发送消息，需要设置不同的instanceName
             原因如下：如果不设置instanceName，那么会使用ip@pid作为producer唯一标识，那么会导致多个producer内部只有一个MQClientInstance(与mq交互)实例，从而导致只往一个集群发消息。
            */
        if (instanceName != null && !"".equals(instanceName)) {
            producer.setInstanceName(instanceName);
        }
        if (producerGroup != null && !"".equals(producerGroup)) {
            producer.setProducerGroup(producerGroup);
        }
        try {
            producer.start();
        } catch (MQClientException e) {
            logger.error(e.getMessage());
        }
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
            }
        }));
    }

    public MqResponse syncSend(String topic, String tags, int flag, String msgId, byte[] body) {
        Message msg = buildMessage(topic, tags, flag, msgId, body);
        SendResult sendResult = null;
        MqResponse mqResponse = new MqResponse();
        mqResponse.setMqSendStatus(MqSendStatus.FAIL);
        try {
            sendResult = producer.send(msg);//send方法支持内部重试
            if (sendResult == null) {
                return mqResponse;
            }
            SendStatus sendStatus = sendResult.getSendStatus();
            if (sendStatus.equals(SendStatus.SEND_OK)) {
                mqResponse.setMqSendStatus(MqSendStatus.SEND_OK);
            }
            logger.info("send message to " + msg.getTopic() + " tags: " + msg.getTags() + ", keys: " + msg.getKeys() + ", sendResult : " + com.alibaba.fastjson.JSON.toJSONString(sendResult));
        } catch (Exception e) {
            e.printStackTrace();
            //如果消息不可丢,必须重发,或者存储到Cache/Db中,由定时器调用重发.
            logger.error("send message to " + msg.getTopic() + " key: " + msg.getKeys() +
                    ", sendResult: " + com.alibaba.fastjson.JSON.toJSONString(sendResult) +
                    ", sendException: " + e.getMessage());
        }
        return mqResponse;
    }

    public void asyncSend(String topic, String tags, int flag, String msgId, byte[] body,
                          final SendCallbackListener sendCallbackListener) {
        Message msg = buildMessage(topic, tags, flag, msgId, body);
        try {
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onSuccess(JSON.toJSONString(sendResult));
                }

                @Override
                public void onException(Throwable e) {
                    logger.error(e.getMessage());
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onException(e);
                }
            });
            logger.info("send message to topic: " + msg.getTopic() + " tags: " + msg.getTags() + ", keys: " + msg.getKeys());
        } catch (Exception e) {
            sendCallbackListener.onException(e);
            e.printStackTrace();
            logger.info("send message to topic: " + msg.getTopic() + " tags: " + msg.getTags() + ", keys: " + msg.getKeys() + ", sendException : " + e.getMessage());
        }
    }

    private Message buildMessage(String topic, String tags, int flag, String msgId, byte[] body) {
        Message msg = new Message();
        msg.setFlag(flag);
        msg.setTopic(topic);
        if (tags != null && !"".equals(tags)) { //一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。
            msg.setTags(tags);
        }
        msg.setKeys(msgId); /*每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突
                             * 如果一个消息包含key值的话，会使用IndexFile存储消息索引，根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目，
                             * 根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)，遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)
                             */
        msg.setBody(body);
        return msg;
    }

    public void shutdown() {
        producer.shutdown();
    }

    public abstract String getProducerGroup();

    public abstract String getNameServerAddress();

    public abstract String getInstanceName();

    private DefaultMQProducer getProducer() {
        if (producer == null) {
            producer = new DefaultMQProducer(getProducerGroup());
        }
        return producer;
    }
}
